30 데이터 수집&전처리&데이터베이스에 삽입 사이클 | ✅ 저자: 이유정(박사)

수집 → staging 테이블에 넣고 → 분산 처리하는 방식을 위한 테이블 추가

테이블 생성하기: 우선 레퍼런스 대상 테이블 먼저 생성한다.

CREATE TABLE restaurants_kakao (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255),
    area_name VARCHAR(100),
    kakao_place_id VARCHAR(100),
    address TEXT,
    phone VARCHAR(20),
    cuisine_type VARCHAR(20),
    category VARCHAR(20),
    tags TEXT,
    menu TEXT,
    rating FLOAT,
    keyword TEXT,
    rating_count INT,
    business_hour VARCHAR(100),
    latitude VARCHAR(20),
    longitude VARCHAR(20),
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

테이블을 실행하고 새로고침을 눌러서 테이블이 생성되는지 확인한다.

id : 각 행(row)을 고유하게 식별하기 위한 기본 키 BIGINT : 정수형 타입으로, INT보다 더 큰 숫자를 저장할 수 있어요. - INT는 약 ±21억까지 - BIGINT는 ±922경까지 저장 가능

• 많은 데이터를 다룰 경우 BIGINT를 사용하면 안전합니다. AUTO_INCREMENT : 값을 자동으로 증가시켜주는 기능입니다.
• 레코드(행)를 삽입할 때 id 값을 입력하지 않아도 1, 2, 3, 4... 이렇게 자동으로 늘어나요. 예:

INSERT INTO restaurants_kakao (name, address) VALUES ('식당A', '서울시...'); 
-- 자동으로 id=1이 들어감

PRIMARY KEY : 이 컬럼을 테이블의 기본 키(Primary Key)로 지정합니다.

• 즉, 중복될 수 없고, 항상 고유(unique)해야 합니다.

전체 의미 요약:

id BIGINT AUTO_INCREMENT PRIMARY KEY

이 컬럼은:

• BIGINT 타입의 큰 정수를 저장하고
• 값을 자동으로 증가시키며
• 고유한 기본키로 테이블에서 각 행을 식별하는 역할을 합니다.

---

created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP

==created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP==

• 이 컬럼은 레코드가 처음 생성될 때의 시간을 저장해요.
• NOT NULL: 비워둘 수 없다는 의미.
• DEFAULT CURRENT_TIMESTAMP: 값을 따로 지정하지 않아도 현재 시각이 자동 입력됨. 예를 들어:

INSERT INTO restaurants_kakao (name) VALUES ('식당A'); 
-- created_at에는 자동으로 현재 시간이 입력됨 (예: 2025-07-20 17:22:35)

==updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP==

• 이 컬럼은 레코드가 수정(update)될 때마다 자동으로 현재 시간으로 갱신돼요.
• DEFAULT CURRENT_TIMESTAMP: 최초 INSERT 시 기본값으로 현재 시각 입력
• ON UPDATE CURRENT_TIMESTAMP: 나중에 UPDATE 될 경우 자동으로 현재 시각으로 갱신

UPDATE restaurants_kakao SET name = '식당B' WHERE id = 1; 
-- updated_at 컬럼이 현재 시각으로 자동 갱신됨

---

생성된 테이블 확인하기 단축키 F5

종속 테이블 만들기

CREATE TABLE restaurant_blog_reviews_kakao (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    title VARCHAR(255),
    published_date DATE,
    blog_url VARCHAR(255),
    restaurant_id BIGINT,
    created_at DATETIME,
    updated_at DATETIME,
    FOREIGN KEY (restaurant_id) REFERENCES restaurants_kakao(id)
);

마지막 테이블 생성

CREATE TABLE restaurant_reviews_kakao (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    title VARCHAR(255),
    author VARCHAR(100),
    content TEXT,
    restaurant_id BIGINT,
    image_urls TEXT,
    created_at DATETIME,
    updated_at DATETIME,
    FOREIGN KEY (restaurant_id) REFERENCES restaurants_kakao(id)
);

---

레스토랑 데이터 크롤러 하기

from time import sleep
import googlemaps
from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.support.ui import WebDriverWait
from sqlalchemy import create_engine
from webdriver_manager.chrome import ChromeDriverManager

# from api_scrap.geocoding import geocode_address
def get_webdriver():
    options = Options()
    options.add_argument('--start-maximized')
    options.add_argument('--disable-blink-features=AutomationControlled')
    options.add_argument('--disable-gpu')
    # options.add_argument('--headless')  
    # 주피터에서 디버깅 중이면 headless 끄세요

    service = Service(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=options)
    return driver

def wait_until_visible(driver, by, value, timeout=10):
    wait = WebDriverWait(driver, timeout)
    wait.until(ec.visibility_of_element_located((by, value)))

def wait_until_clickable(driver, by, value, timeout=10):
    wait = WebDriverWait(driver, timeout)
    wait.until(ec.element_to_be_clickable((by, value)))

def input_and_send(driver, by, value, keys):
    field = driver.find_element(by, value)
    field.send_keys(keys)
    field.send_keys(Keys.ENTER)

def click_button(driver, by, value):
    button = driver.find_element(by, value)
    button.click()

# ------------------ 웹 드라이브 관련부분


def geocode_address(address):
    # Google Maps API 클라이언트를 초기화합니다.
    # '본인의 구글API키 삽입' 부분에 실제 발급받은 API 키를 입력하세요.
    gmaps = googlemaps.Client(key='본인의 구글API키 삽입')
    
    # 주소를 위도(latitude)와 경도(longitude)로 변환하기 위해 지오코딩 요청을 보냅니다.
    geocode_result = gmaps.geocode(address)

    # 결과가 있을 경우
    if geocode_result:
        # 결과 중 첫 번째 항목에서 위치 정보를 추출합니다.
        location = geocode_result[0]['geometry']['location']

        # 위도(latitude)와 경도(longitude)를 따로 저장합니다.
        latitude = location['lat']
        longitude = location['lng']

        # 위도와 경도를 반환합니다.
        return latitude, longitude
    else:
        # 주소에 대한 정보를 찾지 못한 경우 None을 반환합니다.
        return None
        
# ------------------------

def get_items(html: str, parsed_items: list, cuisine: str, category: str):
    # HTML 문자열을 파싱하여 BeautifulSoup 객체 생성
    soup = BeautifulSoup(html, "html.parser")

    # 가게 정보를 담고 있는 리스트 아이템(li 태그)을 모두 선택
    items = soup.select("li.PlaceItem.clickArea")

    # 각 가게 아이템을 순회하며 정보 추출
    for item in items:
        # 주소 정보가 없는 경우는 제외 (예외 처리)
        if item.find('p', {'data-id': 'address'}) is None:
            continue

        # 주소 텍스트를 추출하여 geocode_address 함수로 위도/경도 변환
        loc = geocode_address(item.find('p', {'data-id': 'address'}).text)

        # 위도/경도 값이 존재하면 분리 저장, 아니면 None 처리
        if loc:
            lat, long = loc
        else:
            lat, long = None, None

        # 상세페이지 링크에서 place_id 추출 (URL 마지막 부분)
        more_info = item.find("a", {"data-id": "moreview"})
        place_id = more_info["href"].split("/")[-1]

        # 파싱된 정보를 딕셔너리로 정리하여 parsed_items 리스트에 추가
        parsed_items.append({
            # 식당 이름
            "name": item.find('span', {'data-id': 'screenOutName'}).text,

            # 고유 장소 ID
            "place_id": place_id,

            # 주소
            "address": item.find('p', {'data-id': 'address'}).text,

            # 전화번호
            "phone": item.find('span', {'data-id': 'phone'}).text,

            # 음식 종류 (예: 한식, 중식 등)
            "cuisine_type": cuisine,

            # 카테고리 (예: 맛집, 카페 등)
            "category": category,

            # 평점
            "rating": item.find('em', {'data-id': 'scoreNum'}).text,

            # 영업 시간
            "business_hour": item.find('a', {'data-id': 'periodTxt'}).text,

            # 위도
            "latitude": lat,

            # 경도
            "longitude": long
        })

# ------------------------

def get_sql_connection():
# MySQL 데이터베이스에 연결하기 위한 엔진 생성 및 연결 반환
    engine = create_engine('mysql+pymysql://root:DjangoUserPass!123@localhost:3306/restaurant_db?charset=utf8') # mysql비번과 db가 맞는지 확인
    return engine.connect()

def get_data_from_kakaomap(search_keyword: str, cuisine: str, category: str):
    # 웹드라이버 실행 (Chrome 브라우저 제어용)
    driver = get_webdriver()
    
    # MySQL DB 연결
    db_conn = get_sql_connection()

    try:
        # 카카오맵 메인 페이지 접속
        driver.get("https://map.kakao.com/")
        wait_until_visible(driver, By.ID, "search.keyword.query")         # 검색창 로딩 대기

        # 검색어 입력 후 엔터키 두 번 (검색 실행)
        input_and_send(driver, By.ID, "search.keyword.query", search_keyword)
        input_and_send(driver, By.ID, "search.keyword.query", Keys.ENTER)

        wait_until_clickable(driver, By.ID, "info.search.place.more")  # '더보기' 버튼 클릭 가능할 때까지 대기

        # 간혹 화면을 가리는 어두운 dimmedLayer가 있으면 숨김 처리
        driver.execute_script("""
            var element = document.getElementById('dimmedLayer')
            if (element) {
                element.className = 'DimmedLayer HIDDEN';
            }
        """)

        # 검색 결과 '더보기' 버튼 클릭
        click_button(driver, By.ID, "info.search.place.more")

        wait_until_visible(driver, By.ID, "info.search.page")  # 페이지 네비게이션 영역 로딩 대기

        page_count = 0        # 현재까지 확인한 페이지 수
        items = []            # 크롤링된 아이템들을 저장할 리스트

        # 최대 5페이지까지 반복
        while page_count <= 5:
            page_count += 1
            page_num = page_count % 5 if page_count % 5 != 0 else 5  # 페이지 번호 계산

            # 해당 페이지 버튼 클릭
            click_button(driver, By.ID, f"info.search.page.no{page_num}")
            wait_until_visible(driver, By.ID, "info.search.place.list")  # 리스트 영역 로딩 대기

            # 현재 페이지의 HTML 추출
            place_list = driver.find_element(By.ID, "info.search.place.list")
            shop_list = place_list.get_attribute("innerHTML")

            # HTML 파싱 및 맛집 정보 추출
            get_items(shop_list, items, cuisine, category)

            sleep(1)  # 너무 빠른 요청 방지용

        driver.quit()  
        # 브라우저 종료 (중간 오류가 없다면 여기서 한번 닫힘)

        import pandas as pd

  # 기존 DB에서 해당 음식 종류, 카테고리의 name + address 정보 불러오기
        sql_df = pd.read_sql(
            f'''
            SELECT name, address
            FROM restaurants_kakao
            WHERE cuisine_type="{cuisine}" AND category="{category}"
            ''',
            db_conn
        )

        # 새로 크롤링한 데이터를 DataFrame으로 변환
        df = pd.DataFrame(items)

        # place_id → kakao_place_id 로 컬럼명 변경
        df.rename(columns={'place_id': 'kakao_place_id'}, inplace=True)

        # 기존 DB에 없는 새로운 (name, address) 조합만 필터링
        filtered_df = df[~df[['name', 'address']].apply(tuple, axis=1).isin(
            sql_df[['name', 'address']].apply(tuple, axis=1)
        )]

        # 평점(rating)을 숫자로 변환 (NaN 처리도 허용)
        filtered_df["rating"] = pd.to_numeric(filtered_df["rating"], errors='coerce')

        # 확인용 출력
        print("✔ 저장될 row count:", len(filtered_df))
        print(filtered_df.head())

        # DB에 restaurants_kakao 테이블로 저장 (append 모드)
        filtered_df.to_sql("restaurants_kakao", db_conn, if_exists='append', index=False)

        # to_sql은 자동 commit됨

        return items  # 수집된 전체 아이템 리스트 반환

    except Exception as e:
        # 예외 발생 시 출력 및 에러 전달
        print(e)
        raise e

    finally:
        # 무조건 리소스 정리
        db_conn.close()    # DB 연결 종료
        driver.quit()    # 브라우저 종료 (예외로 종료된 경우에도 대비)

Jupyter book에서 실행하여 크롤링 하기:

import sys
import os
print(os.getcwd())
sys.path.append(os.getcwd())

from crawler.restuarant_crawler import get_data_from_kakaomap

# get_data_from_kakaomap("강남구 카페", "디저트", "카페")
# get_data_from_kakaomap("강남구 해장국", "한식", "국밥")

get_data_from_kakaomap("강남구 치킨", "한식", "치킨")

# search = [
#     ("강남구 파스타", "양식", "파스타"),
#     ("강남구 초밥", "일식", "초밥"),
#     ("강남구 라멘", "일식", "리멘"),
#     ("강남구 스테이크", "양식", "스테이크"),
#     ("강남구 중국집", "중식", "중식"),
#     ("강남구 해물찜", "한식", "찜"),
#     ("강남구 맥주집", "주점", "주점"),
#     ("강남구 족발", "한식", "족발"),
#     ("강남구 와플", "디저트", "디저트"),
#     ("강남구 떡볶이", "한식", "분식"),
    
# ]

# for s in search:
#     get_data_from_kakaomap(*s)

---